package com.shujia.core

import com.shujia.core.Demo09GroupBy.Stu
import com.shujia.core.Demo10GroupByKey.Stu
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

object Demo11ReduceByKey {
  def main(args: Array[String]): Unit = {
    /**
     * reduceByKey：转换算子 需要接收一个聚合函数
     * 首先只有KV格式的RDD才能调用reduceByKey算子
     * 可以将KV格式的RDD按照Key进行分组并且同时进行聚合操作
     * 聚合操作需要通过函数进行传入
     * 同时也会将聚合函数用于预聚合
     */

    val conf: SparkConf = new SparkConf()
    conf.setMaster("local")
    conf.setAppName("Demo11ReduceByKey")

    val sc: SparkContext = new SparkContext(conf)

    // 按照班级分组并且统计班级人数
    val stuKVRDD: RDD[(String, Int)] = sc
      .textFile("spark/data/stu/students.txt")
      // 将数据变成KV格式，以班级作为Key，1作为Value
      .map(line => {
        val splits: Array[String] = line.split(",")
        (splits(4), 1)
      })

    stuKVRDD
      .groupByKey()
      .map(kv => {
        val clazz: String = kv._1
        //        val cnt: Int = kv._2.size
        val cnt: Int = kv._2.sum
        s"$clazz,$cnt"
      }).foreach(println)

    // 使用reduceByKey进行简化
    stuKVRDD
      // reduceByKey无法直接完成avg聚合操作
      // reduceByKey 可以实现预聚合操作
      //      .reduceByKey((i1: Int, i2: Int) => i1 + i2)
      .reduceByKey(_ + _) // 基于匿名函数的省略规则进行简化
      .map(kv => s"${kv._1},${kv._2}")
      .foreach(println)

    // 使用aggregateByKey统计班级的平均年龄
    //    stuKVRDD.aggregateByKey()



  }

}
